/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.seatunnel.connectors.seatunnel.file.source.reader;

import org.apache.seatunnel.api.common.SeaTunnelAPIErrorCode;
import org.apache.seatunnel.api.configuration.ReadonlyConfig;
import org.apache.seatunnel.api.serialization.DeserializationSchema;
import org.apache.seatunnel.api.source.Collector;
import org.apache.seatunnel.api.table.catalog.CatalogTable;
import org.apache.seatunnel.api.table.catalog.CatalogTableUtil;
import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
import org.apache.seatunnel.api.table.type.SeaTunnelRow;
import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
import org.apache.seatunnel.common.utils.DateTimeUtils;
import org.apache.seatunnel.common.utils.DateUtils;
import org.apache.seatunnel.common.utils.TimeUtils;
import org.apache.seatunnel.connectors.seatunnel.file.config.CompressFormat;
import org.apache.seatunnel.connectors.seatunnel.file.config.FileBaseSourceOptions;
import org.apache.seatunnel.connectors.seatunnel.file.config.FileFormat;
import org.apache.seatunnel.connectors.seatunnel.file.exception.FileConnectorErrorCode;
import org.apache.seatunnel.connectors.seatunnel.file.exception.FileConnectorException;
import org.apache.seatunnel.format.text.TextDeserializationSchema;
import org.apache.seatunnel.format.text.constant.TextFormatConstant;
import org.apache.seatunnel.format.text.splitor.DefaultTextLineSplitor;
import org.apache.seatunnel.format.text.splitor.TextLineSplitor;

import io.airlift.compress.lzo.LzopCodec;
import lombok.extern.slf4j.Slf4j;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;
import java.util.Map;
import java.util.Optional;

@Slf4j
public class TextReadStrategy extends AbstractReadStrategy {
    private DeserializationSchema<SeaTunnelRow> deserializationSchema;
    private String fieldDelimiter = FileBaseSourceOptions.FIELD_DELIMITER.defaultValue();
    private String rowDelimiter = FileBaseSourceOptions.ROW_DELIMITER.defaultValue();
    private DateUtils.Formatter dateFormat =
            FileBaseSourceOptions.DATE_FORMAT_LEGACY.defaultValue();
    private DateTimeUtils.Formatter datetimeFormat =
            FileBaseSourceOptions.DATETIME_FORMAT_LEGACY.defaultValue();
    private TimeUtils.Formatter timeFormat =
            FileBaseSourceOptions.TIME_FORMAT_LEGACY.defaultValue();
    private CompressFormat compressFormat = FileBaseSourceOptions.COMPRESS_CODEC.defaultValue();
    private TextLineSplitor textLineSplitor;
    private int[] indexes;
    private String encoding = FileBaseSourceOptions.ENCODING.defaultValue();

    /** Custom stream divider for splitting text streams by specified delimiters */
    public static class StreamLineSplitter {
        private final char[] delimiterChars;
        private final StringBuilder lineBuffer;
        private int delimiterIndex;
        private int skipCount;
        private final long skipHeaderNumber;
        private final LineProcessor lineProcessor;
        private final boolean useReadLine;

        public StreamLineSplitter(
                String delimiter, long skipHeaderNumber, LineProcessor lineProcessor) {
            this.delimiterChars = delimiter.toCharArray();
            this.lineBuffer = new StringBuilder();
            this.delimiterIndex = 0;
            this.skipCount = 0;
            this.skipHeaderNumber = skipHeaderNumber;
            this.lineProcessor = lineProcessor;

            this.useReadLine = isDefaultLineDelimiter(delimiter);
        }

        private boolean isDefaultLineDelimiter(String delimiter) {
            return "\n".equals(delimiter) || "\r".equals(delimiter) || "\r\n".equals(delimiter);
        }

        public void processStream(BufferedReader reader) throws IOException {
            if (useReadLine) {
                processWithReadLine(reader);
            } else {
                processWithCharByChar(reader);
            }
        }

        private void processWithReadLine(BufferedReader reader) throws IOException {
            String line;
            int lineCount = 0;

            while ((line = reader.readLine()) != null) {
                if (lineCount >= skipHeaderNumber) {
                    if (!line.trim().isEmpty()) {
                        lineProcessor.processLine(line);
                    }
                } else {
                    lineCount++;
                }
            }
        }

        private void processWithCharByChar(BufferedReader reader) throws IOException {
            int ch;
            while ((ch = reader.read()) != -1) {
                char currentChar = (char) ch;
                processChar(currentChar);
            }

            if (lineBuffer.length() > 0) {
                if (skipCount >= skipHeaderNumber) {
                    String line = lineBuffer.toString();
                    if (!line.trim().isEmpty()) {
                        lineProcessor.processLine(line);
                    }
                }
            }
        }

        private void processChar(char currentChar) throws IOException {
            if (currentChar == delimiterChars[delimiterIndex]) {
                delimiterIndex++;
                if (delimiterIndex == delimiterChars.length) {
                    if (skipCount >= skipHeaderNumber) {
                        String line = lineBuffer.toString();
                        if (!line.trim().isEmpty()) {
                            lineProcessor.processLine(line);
                        }
                    } else {
                        skipCount++;
                    }

                    lineBuffer.setLength(0);
                    delimiterIndex = 0;
                }
            } else {
                if (delimiterIndex > 0) {
                    for (int i = 0; i < delimiterIndex; i++) {
                        lineBuffer.append(delimiterChars[i]);
                    }
                    delimiterIndex = 0;
                }
                lineBuffer.append(currentChar);
            }
        }
    }

    public interface LineProcessor {
        void processLine(String line) throws IOException;
    }

    @Override
    public void read(String path, String tableId, Collector<SeaTunnelRow> output)
            throws FileConnectorException, IOException {
        Map<String, String> partitionsMap = parsePartitionsByPath(path);
        resolveArchiveCompressedInputStream(path, tableId, output, partitionsMap, FileFormat.TEXT);
    }

    @Override
    public void readProcess(
            String path,
            String tableId,
            Collector<SeaTunnelRow> output,
            InputStream inputStream,
            Map<String, String> partitionsMap,
            String currentFileName)
            throws IOException {
        InputStream actualInputStream;
        switch (compressFormat) {
            case LZO:
                LzopCodec lzo = new LzopCodec();
                actualInputStream = lzo.createInputStream(inputStream);
                break;
            case NONE:
                actualInputStream = inputStream;
                break;
            default:
                log.warn(
                        "Text file does not support this compress type: {}",
                        compressFormat.getCompressCodec());
                actualInputStream = inputStream;
                break;
        }

        try (BufferedReader reader =
                new BufferedReader(new InputStreamReader(actualInputStream, encoding))) {

            LineProcessor lineProcessor =
                    line -> {
                        try {
                            processLineData(line, tableId, output, partitionsMap);
                        } catch (FileConnectorException e) {
                            throw new IOException(e);
                        }
                    };

            StreamLineSplitter splitter =
                    new StreamLineSplitter(rowDelimiter, skipHeaderNumber, lineProcessor);
            splitter.processStream(reader);
        }
    }

    private void processLineData(
            String line,
            String tableId,
            Collector<SeaTunnelRow> output,
            Map<String, String> partitionsMap)
            throws FileConnectorException {
        try {
            SeaTunnelRow seaTunnelRow =
                    deserializationSchema.deserialize(line.getBytes(StandardCharsets.UTF_8));
            if (!readColumns.isEmpty()) {
                // need column projection
                Object[] fields;
                if (isMergePartition) {
                    fields = new Object[readColumns.size() + partitionsMap.size()];
                } else {
                    fields = new Object[readColumns.size()];
                }
                for (int i = 0; i < indexes.length; i++) {
                    fields[i] = seaTunnelRow.getField(indexes[i]);
                }
                seaTunnelRow = new SeaTunnelRow(fields);
            }
            if (isMergePartition) {
                int index = seaTunnelRowType.getTotalFields();
                for (String value : partitionsMap.values()) {
                    seaTunnelRow.setField(index++, value);
                }
            }
            seaTunnelRow.setTableId(tableId);
            output.collect(seaTunnelRow);
        } catch (IOException e) {
            String errorMsg =
                    String.format(
                            "Deserialize this data [%s] failed, please check the origin data",
                            line);
            throw new FileConnectorException(
                    FileConnectorErrorCode.DATA_DESERIALIZE_FAILED, errorMsg, e);
        }
    }

    @Override
    public SeaTunnelRowType getSeaTunnelRowTypeInfo(String path) {
        this.seaTunnelRowType = CatalogTableUtil.buildSimpleTextSchema();
        this.seaTunnelRowTypeWithPartition =
                mergePartitionTypes(fileNames.get(0), seaTunnelRowType);
        initFormatter();
        if (pluginConfig.hasPath(FileBaseSourceOptions.READ_COLUMNS.key())) {
            throw new FileConnectorException(
                    SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED,
                    "When reading text files, if user has not specified schema information, "
                            + "SeaTunnel will not support column projection");
        }
        ReadonlyConfig readonlyConfig = ReadonlyConfig.fromConfig(pluginConfig);
        TextDeserializationSchema.Builder builder =
                TextDeserializationSchema.builder()
                        .delimiter(TextFormatConstant.PLACEHOLDER)
                        .textLineSplitor(textLineSplitor)
                        .nullFormat(
                                readonlyConfig
                                        .getOptional(FileBaseSourceOptions.NULL_FORMAT)
                                        .orElse(null));
        if (isMergePartition) {
            deserializationSchema =
                    builder.seaTunnelRowType(this.seaTunnelRowTypeWithPartition).build();
        } else {
            deserializationSchema = builder.seaTunnelRowType(this.seaTunnelRowType).build();
        }
        return getActualSeaTunnelRowTypeInfo();
    }

    @Override
    public void setCatalogTable(CatalogTable catalogTable) {
        SeaTunnelRowType rowType = catalogTable.getSeaTunnelRowType();
        SeaTunnelRowType userDefinedRowTypeWithPartition =
                mergePartitionTypes(fileNames.get(0), rowType);
        ReadonlyConfig readonlyConfig = ReadonlyConfig.fromConfig(pluginConfig);
        Optional<String> fieldDelimiterOptional =
                readonlyConfig.getOptional(FileBaseSourceOptions.FIELD_DELIMITER);
        Optional<String> rowDelimiterOptional =
                readonlyConfig.getOptional(FileBaseSourceOptions.ROW_DELIMITER);
        encoding =
                readonlyConfig
                        .getOptional(FileBaseSourceOptions.ENCODING)
                        .orElse(StandardCharsets.UTF_8.name());
        fieldDelimiterOptional.ifPresent(s -> fieldDelimiter = s);
        rowDelimiterOptional.ifPresent(s -> rowDelimiter = s);
        initFormatter();
        TextDeserializationSchema.Builder builder =
                TextDeserializationSchema.builder()
                        .delimiter(fieldDelimiter)
                        .textLineSplitor(textLineSplitor)
                        .nullFormat(
                                readonlyConfig
                                        .getOptional(FileBaseSourceOptions.NULL_FORMAT)
                                        .orElse(null));
        if (isMergePartition) {
            deserializationSchema =
                    builder.seaTunnelRowType(userDefinedRowTypeWithPartition).build();
        } else {
            deserializationSchema = builder.seaTunnelRowType(rowType).build();
        }
        // column projection
        if (pluginConfig.hasPath(FileBaseSourceOptions.READ_COLUMNS.key())) {
            // get the read column index from user-defined row type
            indexes = new int[readColumns.size()];
            String[] fields = new String[readColumns.size()];
            SeaTunnelDataType<?>[] types = new SeaTunnelDataType[readColumns.size()];
            for (int i = 0; i < indexes.length; i++) {
                indexes[i] = rowType.indexOf(readColumns.get(i));
                fields[i] = rowType.getFieldName(indexes[i]);
                types[i] = rowType.getFieldType(indexes[i]);
            }
            this.seaTunnelRowType = new SeaTunnelRowType(fields, types);
            this.seaTunnelRowTypeWithPartition =
                    mergePartitionTypes(fileNames.get(0), this.seaTunnelRowType);
        } else {
            this.seaTunnelRowType = rowType;
            this.seaTunnelRowTypeWithPartition = userDefinedRowTypeWithPartition;
        }
    }

    private void initFormatter() {
        if (pluginConfig.hasPath(FileBaseSourceOptions.DATE_FORMAT_LEGACY.key())) {
            dateFormat =
                    DateUtils.Formatter.parse(
                            pluginConfig.getString(FileBaseSourceOptions.DATE_FORMAT_LEGACY.key()));
        }
        if (pluginConfig.hasPath(FileBaseSourceOptions.DATETIME_FORMAT_LEGACY.key())) {
            datetimeFormat =
                    DateTimeUtils.Formatter.parse(
                            pluginConfig.getString(
                                    FileBaseSourceOptions.DATETIME_FORMAT_LEGACY.key()));
        }
        if (pluginConfig.hasPath(FileBaseSourceOptions.TIME_FORMAT_LEGACY.key())) {
            timeFormat =
                    TimeUtils.Formatter.parse(
                            pluginConfig.getString(FileBaseSourceOptions.TIME_FORMAT_LEGACY.key()));
        }
        if (pluginConfig.hasPath(FileBaseSourceOptions.COMPRESS_CODEC.key())) {
            String compressCodec =
                    pluginConfig.getString(FileBaseSourceOptions.COMPRESS_CODEC.key());
            compressFormat = CompressFormat.valueOf(compressCodec.toUpperCase());
        }
        textLineSplitor = new DefaultTextLineSplitor();
    }
}
